package com.bfxy.rabbit.producer.broker;

import com.bfxy.rabbit.api.enums.MessageTypeEnum;
import com.bfxy.rabbit.api.exception.MessageRunTimeException;
import com.bfxy.rabbit.api.model.Message;
import com.bfxy.rabbit.common.convert.GenericMessageConverter;
import com.bfxy.rabbit.common.convert.RabbitMessageConvert;
import com.bfxy.rabbit.common.serializer.Serializer;
import com.bfxy.rabbit.common.serializer.SerializerFactory;
import com.bfxy.rabbit.common.serializer.impl.JacksonSerializerFactory;
import com.bfxy.rabbit.producer.enums.BrokerMessageStatusEnum;
import com.bfxy.rabbit.producer.service.MessageStoreService;
import com.google.common.base.Preconditions;
import com.google.common.base.Splitter;
import com.google.common.collect.Maps;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.retry.support.RetryTemplate;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.List;
import java.util.Map;

/**
 * $RabbitTemplateContainer RabbitTemplate池化封装
 * 每一个topic，对应一个RabbitTemplate
 * 1.提高发送的效率
 * 2.可以根据不用的需求定制化不同的RabbitTemplate，比如每一个topic都有自己的routingKey规则
 */
@Slf4j
@Component
public class RabbitTemplateContainer implements RabbitTemplate.ConfirmCallback {

    private Map<String/* TOPIC */, RabbitTemplate> rabbitMap = Maps.newConcurrentMap();

    private final Splitter splitter = Splitter.on("#");

    private final SerializerFactory serializerFactory = JacksonSerializerFactory.INSTANCE;

    @Resource
    private ConnectionFactory connectionFactory;

    @Resource
    private MessageStoreService messageStoreService;

    public RabbitTemplate getRabbitTemplate(Message message) throws MessageRunTimeException {
        Preconditions.checkNotNull(message);
        String topic = message.getTopic();
        RabbitTemplate rabbitTemplate = rabbitMap.get(topic);
        if (rabbitTemplate != null) {
            return rabbitTemplate;
        }
        log.info("#RabbitTemplateContainer.getRabbitTemplate# topic: {} is not exists, create one", topic);
        RabbitTemplate newRabbitTemplate = new RabbitTemplate(connectionFactory);
        newRabbitTemplate.setExchange(topic);
        newRabbitTemplate.setRoutingKey(message.getRoutingKey());
        newRabbitTemplate.setRetryTemplate(new RetryTemplate());

        // 对于message添加序列化和反序列化方式
        Serializer serializer = serializerFactory.create();
        GenericMessageConverter genericMessageConverter = new GenericMessageConverter(serializer);
        RabbitMessageConvert rabbitMessageConvert = new RabbitMessageConvert(genericMessageConverter);
        newRabbitTemplate.setMessageConverter(rabbitMessageConvert);

        MessageTypeEnum messageType = message.getMessageType();
        // 只要不是迅速消息，都应该有消息的confirm确认
        if (!MessageTypeEnum.RAPID.equals(messageType)) {
            newRabbitTemplate.setConfirmCallback(this);
        }

        rabbitMap.putIfAbsent(topic, newRabbitTemplate);

        return rabbitMap.get(topic);
    }

    /**
     * 无论是confirm消息还是reliant消息，发送消息以后broker都会去回调confirm。但是confirm消息不需要入库，而reliant消息需要，故消息重发使用reliant消息
     * @param correlationData
     * @param ack
     * @param cause
     */
    @Override
    public void confirm(CorrelationData correlationData, boolean ack, String cause) {
        // 具体的消息应答
        if (correlationData != null && StringUtils.isNotBlank(correlationData.getId())) {
            List<String> strings = splitter.splitToList(correlationData.getId());
            String messageId = strings.get(0);
            long sendTime = Long.parseLong(strings.get(1));
            String messageType = strings.get(2);
            if (ack) {
                // 当broker返回ack成功时，更新下日志表里对应消息发送状态为SEND_OK
                // 注意：如果当前消息类型为reliant，才去数据库查找并更新
                if (MessageTypeEnum.RELIANT.getValue().equals(messageType)) {
                    messageStoreService.updateStatusByMessageId(messageId, BrokerMessageStatusEnum.SEND_OK);
                }
                log.info("#RabbitTemplateContainer.confirm# send message is OK, confirm messageId: {}, sendTime: {}", messageId, sendTime);
            } else {
                log.error("#RabbitTemplateContainer.confirm# send message is Fail, confirm messageId: {}, sendTime: {}", messageId, sendTime);
            }
        }
    }

}
